package me.tuyou.stream.window;

import me.tuyou.common.LineSplitter;
import me.tuyou.common.WordStreamGenerator;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * 单词统计(流处理)
 */
public class TumblingWindowTest {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<Tuple2<String, Integer>> dataStream = env.addSource(new WordStreamGenerator())
                .flatMap(new LineSplitter())
                .keyBy(0)
//                .window(TumblingProcessingTimeWindows.of(Time.seconds(3)))
                .window(SlidingProcessingTimeWindows.of(Time.seconds(3), Time.seconds(1)))
                .reduce(new ReduceFunction<Tuple2<String, Integer>>() {
                    @Override
                    public Tuple2<String, Integer> reduce(Tuple2<String, Integer> v1, Tuple2<String, Integer> v2) throws Exception {
                        return new Tuple2<>(v1.f0, v1.f1 + v2.f1);
                    }
                }, new ProcessWindowFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple, TimeWindow>() {
                    @Override
                    public void process(Tuple tuple, Context context, Iterable<Tuple2<String, Integer>> iterable, Collector<Tuple2<String, Integer>> collector) throws Exception {
                        for (Tuple2<String, Integer> item : iterable) {
                            collector.collect(item);
                        }
                    }
                });

        dataStream.print();
        env.execute("wordCount");
    }
}
